<?php
apf_require_class("APF_DB_Factory");
abstract class Oa_Job_Abstract {

    const MAX_PAGE = 10000;

    const PER_PAGE = 500;

    private $_pdo = array();

    public $pdo_name = 'nhstats'; //jyang


    /**
     * cache adapter
     * @var object
     */
    protected $_mem = null;

    /**
     * @var boolean
     */
    public $is_log = true;

    public function __construct() {
        $this->lock_file = "/tmp/" . get_class($this) . ".lock";
    }

    /**
     * get rows from database
     *
     * @param string $sql
     * @param string $adapter
     * @param array $params
     * @param boolean $one
     *
     */
    protected function get_rows($sql, $params = array(), $one = false, $adapter = null) {
        if($one) {
            return $this->get_db_row($sql, $adapter, $params);
        } else {
            return $this->get_db_rows($sql, $adapter, $params);
        }
    }

    protected function execute($sql, $params = array(), $adapter = null) {
        $this->log($sql);
        $pdo = $this->get_pdo($adapter);
        $stmt = $pdo->prepare($sql);
        $stmt->execute($params);
        $id = $pdo->lastInsertId();
        if($id) {
            return $id;
        } else {
            return $stmt->rowCount();
        }
    }

    /**
     * alter database data
     *
     * @param string $sql
     * @param string $adapter
     * @param array $params
     *
     */
    protected function alter_db_data($sql, $adapter, $params = array()) {
        $this->log($sql);
        $pdo = $this->get_pdo($adapter);
        $stmt = $pdo->prepare($sql);
        $stmt->execute($params);
        return $stmt->rowCount();
    }

    /**
     * get row from database
     *
     * @param string $sql
     * @param string $adapter
     * @param array $params
     *
     */
    protected function get_db_row($sql, $adapter, $params = array()) {
        $pdo = $this->get_pdo($adapter);
        $stmt = $pdo->prepare($sql);
        $stmt->execute($params);
        return (array)$stmt->fetch();
    }

    /**
     * get rows from database
     *
     * @param string $sql
     * @param string $adapter
     * @param array $params
     *
     */
    protected function get_db_rows($sql, $adapter, $params = array()) {
        $this->log($sql);
        $pdo = $this->get_pdo($adapter);
        $stmt = $pdo->prepare($sql);
        $stmt->execute($params);
        return $stmt->fetchAll();
    }

    /**
     * 对于比较大的表，分多次取出数据，一次取出几千条数据有可能造成 slow query
     * @param  $sql 不要加 Limit 不要加 order by
     * @param  $adapter
     * @param  $index
     * @param  $params
     * @param  $limit
     */
    protected function get_db_more_rows($sql, $adapter, $index, $params = array()) {
        if(empty($sql) || empty($adapter) || empty($index)) {
            $this->log('get_db_more_rows error params');
            return;
        }

        $rt = array();
        $page = 1;
        while(true) {
            if($page > self::MAX_PAGE) {
                $this->log("TOO many data please check: $sql");
                exit(1);
            }
            $offset = ($page - 1) * self::PER_PAGE;
            $_sql = $sql . " order by $index asc limit $offset," . self::PER_PAGE;
            $query = $this->get_db_rows($_sql, $adapter, $params);
            if(empty($query)) {
                break;
            }
            $rt = array_merge($rt, $query);
            $page++;
        }
        return $rt;
    }

    /**
     *
     * 从表中取出所有的数据
     * @param string $sql
     * @param string $adapter
     * @param string $index
     * @param array $params
     * @param boolean $has_where
     */
    protected function get_db_many_rows($sql, $adapter, $index, $where = '', $limit = self::PER_PAGE) {
        if(empty($sql) || empty($adapter) || empty($index)) {
            $this->log('get_db_more_rows error params');
            return;
        }

        $rt = array();
        $page = 1;
        $max_id = 0;
        while(true) {
            if($page > self::MAX_PAGE) {
                $this->log("TOO many data please check: $sql");
                exit(1);
            }

            $_sql = $sql . " where 1 and $where {$index} > $max_id order by $index asc limit " . $limit;

            $query = $this->get_db_rows($_sql, $adapter, array());
            if(empty($query)) {
                break;
            }
            $rt = array_merge($rt, $query);
            $tmp = array_pop($query);
            $max_id = intval($tmp[$index]);
            $page++;
        }
        return $rt;
    }

    /**
     * 一次查询太多的ID可能有问题，所以分批查询
     * @param string $sql
     * @param array $loupan_ids
     */
    protected function get_db_large_data($sql, $ids, $change_index = 1, $adapter = 'nhslave', $limit = 100) {
        if(empty($ids) || ! is_array($ids)) {
            return array();
        }

        apf_require_class('Aifang_Core_Util_Tools');
        //对大的ID数组进行切割
        $_limit = $limit > 0 ? $limit : self::LIMIT;
        sort($ids);
        $ids = array_unique($ids);
        $loupan_ids_split = array_chunk($ids, $_limit);
        $data = array();
        foreach($loupan_ids_split as $_ids) {
            $this->log("memory:" . (memory_get_usage() / 1000000));
            $_ids_str = Aifang_Core_Util_Tools::join_id_array($_ids);
            $_sql = sprintf($sql, $_ids_str);
            if($change_index) {
                //用loupan_id 作索引
                $query = $this->get_db_rows_assoc($_sql, $adapter);
                $data += $query;
            } else {
                $query = $this->get_db_rows($_sql, $adapter);
                $data = array_merge($data, $query);
            }
        }
        return $data;
    }

    /**
     * 用第一个字段作索引，这个功能挺常用的
     * @param string $sql
     * @param string $adapter eg nhslave
     * @param string $index 主键
     * @param array $params
     */
    protected function get_db_more_rows_assoc($sql, $adapter, $index, $params = array()) {
        $query = $this->get_db_more_rows($sql, $adapter, $index, $params);
        if(empty($query)) {
            return array();
        }

        $rs = array();
        foreach($query as $val) {
            $tmp = array_values(array_slice($val, 0, 1));
            $rs[$tmp[0]] = $val;
        }
        return $rs;
    }

    /**
     *
     *用第一个键值作索引
     */
    protected function get_db_rows_assoc($sql, $adapter, $params = array()) {
        $query = $this->get_db_rows($sql, $adapter, $params);
        if(empty($query)) {
            return array();
        }

        $rs = array();
        foreach($query as $val) {
            $tmp = array_values(array_slice($val, 0, 1));
            $rs[$tmp[0]] = $val;
        }
        return $rs;
    }

    /**
     * log data to stdout
     *
     * @param string $str
     */
    protected function log($str = '') {
        if($this->is_log) {
            echo PHP_EOL . date('Y-m-d H:i:s') . ": " . $str . PHP_EOL;
        }
    }

    /**
     * get database pdo connect adapter
     *
     * @param string $slave adapter name
     *
     * @return APF_DB_PDO
     */
    protected function get_pdo($slave) {
        //        if(empty($slave)){
        //            $slave = $this->pdo_name;
        //        }
        if(! array_key_exists($slave, $this->_pdo)) {
            apf_require_class("APF_DB_Factory");
            $this->_pdo[$slave] = APF_DB_Factory::get_instance()->get_pdo($slave);
        }
        return $this->_pdo[$slave];
    }

    /**
     *
     * @return APF_Cache_Memcache
     */
    protected function get_mem() {
        if(null === $this->_mem) {
            apf_require_class("APF_Cache_Factory");
            $this->_mem = APF_Cache_Factory::get_instance()->get_memcache();
        }
        return $this->_mem;
    }

    protected function get_tag () {
        if (null === $this->_tag) {
            apf_require_class('Aifang_Core_Cache_Tag');
            $this->_tag = Aifang_Core_Cache_Tag::get_instance();
        }
        return $this->_tag;
    }
    /**
     * all valid city id list
     *
     * @return array
     */
    protected function get_city_ids() {
        $this->_city_ids = APF::get_instance()->get_config('aifang_cities', 'multicity');
        return $this->_city_ids;
    }

    public function lock_run($params = array()) {
        if($this->is_lock(8 * 3600)) {
            $this->log('is running exit');
            return false;
        }

        $flag = $this->lock();
        if(! $this->lock()) {
            return false;
        }
        $this->run($params = array());
        $flag = $this->unlock();
        return true;
    }

    protected function is_lock($time) {
        $timestamp = time();
        if(is_file($this->lock_file) && ((filemtime($this->lock_file) + $time) > time())) {
            return true;
        } else {
            return false;
        }
    }

    protected function lock() {
        if(! touch($this->lock_file)) {
            $this->log('can not lock');
            return false;
        } else {
            return true;
        }
    }

    protected function unlock() {
        return unlink($this->lock_file);
    }
}